package com.kepler.social.domain.support.media;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.kepler.social.common.utils.ValidateAssert;
import com.kepler.social.domain.support.media.consts.CallBackType;
import com.kepler.social.domain.support.media.mq.SliceMergeConsumer;
import com.kepler.social.domain.support.media.mq.SliceMergeMessage;
import com.kepler.social.domain.support.media.mq.SliceMergeProducer;
import com.kepler.social.domain.support.media.vo.CallBackVo;
import com.kepler.social.domain.support.media.vo.FileSliceVo;
import com.kepler.social.domain.support.media.vo.M3u8Vo;
import com.lmax.disruptor.dsl.Disruptor;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Path;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class MediaDomainService implements InitializingBean, DisposableBean {

    @Resource
    private RedisTemplate redisTemplate;
    @Resource
    private MediaHelper mediaHelper;

    private ObjectMapper objectMapper = new ObjectMapper();

    private Disruptor<SliceMergeMessage> mergeDisruptor;

    private SliceMergeProducer mergeProducer;


    /**
     * 上传文件分片
     * 给外部调用，体现的是上传
     *
     * @param reqSeq
     * @param stream
     * @param currentPart 当前上传块号 todo 乐观锁设置如redis,保证不会被多次触发
     * @param totalPart   总块数
     * @throws IOException
     */
    public void uploadFileSlice(String reqSeq, InputStream stream, int currentPart, int totalPart) throws IOException {
        int bytesRead;
        int len = 1024;
        byte[] buffer = new byte[len];
        while ((bytesRead = stream.read(buffer, 0, len)) != -1) {
            byte[] tar = new byte[bytesRead];
            System.arraycopy(buffer, 0, tar, 0, bytesRead);
            this.storeSliceDataBlock(reqSeq, tar, 20);
        }
    }

    /**
     * 并发安全保存分片数据
     * 用保存，是因为是内部方法，体现的是具体的操作
     * todo  public ->  private
     *
     * @param reqSeq  请求序列号, 同时也是分布式锁
     * @param bytes   分片数据
     * @param timeOut 暂时存储的过期时间 秒
     * @throws JsonProcessingException
     */
    public void storeSliceDataBlock(String reqSeq, byte[] bytes, int timeOut)
            throws JsonProcessingException {
        redisTemplate.opsForList().rightPush(reqSeq, objectMapper.writeValueAsString(bytes));
        redisTemplate.expire(reqSeq, timeOut, TimeUnit.MINUTES);
    }

    public M3u8Vo mergeVideoSlice(String reqSeq, String originalFileName) throws IOException {
        return mediaHelper.mergeVideoSlice(reqSeq, originalFileName);
    }

    public String mergeImageSlice(String reqSeq, String originalFileName) throws IOException {
        return mediaHelper.mergeImageSlice(reqSeq, originalFileName);
    }

    /**
     * 提交合并分片任务
     *
     * @param fileSlices
     * @param callBackVo
     */
    public void submitMergeSliceTask(List<FileSliceVo> fileSlices, CallBackVo callBackVo) {
        CallBackType callBackType = callBackVo.getCallBackType();
        ValidateAssert.isTrue(callBackType == CallBackType.spring_ioc, "目前只支持Spring ioc");
        /// 先merge
        mergeProducer.send(fileSlices, callBackVo);
    }

    public M3u8Vo storeAsM3U8(String title, Path sourceFilePath) throws IOException {
        return mediaHelper.storeAsM3U8(title, sourceFilePath);
    }

    @Override
    public void afterPropertiesSet() {
        mergeDisruptor = new Disruptor(SliceMergeMessage::new, 1024, Executors.defaultThreadFactory());
        mergeDisruptor.handleEventsWithWorkerPool(new SliceMergeConsumer());
        mergeDisruptor.start();
        mergeProducer = new SliceMergeProducer(mergeDisruptor);
    }

    @Override
    public void destroy() {
        mergeDisruptor.shutdown();
    }

}